package com.winit.cloudlink.event;

/**
 * Created by wanglin on 2018/5/28.
 */
import com.alibaba.fastjson.JSONObject;
import com.winit.cloudlink.config.Metadata;
import com.winit.cloudlink.event.eventid.EventIdFactory;
import com.winit.cloudlink.event.internal.EventMessageHandler;
import com.winit.cloudlink.message.DefaultMessageEngine;
import com.winit.cloudlink.message.Message;
import com.winit.cloudlink.message.MessageBuilder;
import com.winit.cloudlink.message.MessageEngine;
import com.winit.cloudlink.message.MessageEngine.HandlerRegisterCallback;
import com.winit.cloudlink.message.handler.MessageHandler.HandlerType;
import com.winit.common.util.SpringBean;
import com.winit.redis.service.RedisService;

import java.io.Serializable;

public class DefaultEventEngine implements EventEngine {
    private Metadata metadata;
    private MessageEngine messageEngine;

    private RedisService redisService;

    public DefaultEventEngine(Metadata metadata) {
        this(metadata, new DefaultMessageEngine(metadata));
    }

    public DefaultEventEngine(Metadata metadata, MessageEngine messageEngine) {
        this.metadata = metadata;
        this.messageEngine = messageEngine;
        EventIdFactory.me().init(metadata);
    }

    public Metadata getMetadata() {
        return this.metadata;
    }

    public MessageEngine getMessageEngine() {
        return this.messageEngine;
    }

    @Override
    public <Payload extends Serializable> void publishEvent(Event<Payload> event) {

        redisService= (RedisService) SpringBean.getBean("transferService");
        Message<?> message = (new MessageBuilder(this.metadata)).topic(event.getHeaders().getEventName(), event, event.getHeaders().getRoutingKey(), event.getHeaders().getZones()).build();
        JSONObject json =(JSONObject) redisService.getObj(event.getHeaders().getEventId());
        redisService.delObj(event.getHeaders().getEventId());
        redisService.setObj(message.getHeaders().getMessageId(),json);
        this.messageEngine.send(message);
    }

    @Override
    public <Payload extends Serializable> void registerEventListener(EventListener<Event<Payload>> eventListener, final HandlerRegisterCallback callback) {
        EventMessageHandler<Event<Payload>> messageHandler = new EventMessageHandler(eventListener, this, this.metadata);
        this.messageEngine.register(messageHandler, new HandlerRegisterCallback() {
            @Override
            public void onCompleted(HandlerType handlerType, String messageType) {
                if(null != callback) {
                    callback.onCompleted(HandlerType.EVENT, messageType);
                }

            }
        });
    }

    @Override
    public <Payload extends Serializable> void unRegisterEventListener(EventListener<Event<Payload>> eventListener) {
        EventMessageHandler<Event<Payload>> messageHandler = new EventMessageHandler(eventListener, this, this.metadata);
        this.messageEngine.unregister(messageHandler);
    }

    @Override
    public <Param> EventBuilder<Param> newEventBuilder() {
        return new EventBuilder(this.metadata);
    }

    @Override
    public void start() {
        this.messageEngine.start();
    }

    @Override
    public void shutdown() {
        this.messageEngine.shutdown();
    }
}